package club.vann.kafka.mock.kafka;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.function.Consumer;
import lombok.Value;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.util.ObjectUtils;

@Slf4j
@Value
public class KafkaConsumer {
  /** 订阅了kafka消息的consumer集合, key值为topic + key, value为consumer集合 */
  Map<String, Set<Consumer<byte[]>>> consumerMap;

  /** kafka配置信息 */
  private static KafkaProperties kafkaProperties;

  /** kafkaConsumer实例 */
  private static KafkaConsumer kafkaConsumer;

  /** 默认构造方法 */
  public KafkaConsumer() {
    this.consumerMap = new ConcurrentHashMap<>();
  }

  public static KafkaConsumer init(KafkaProperties kafkaProperties, KafkaConsumer kafkaConsumer) {
    KafkaConsumer.kafkaConsumer = kafkaConsumer;
    KafkaConsumer.kafkaProperties = kafkaProperties;
    return kafkaConsumer;
  }

  /** 清空 kafka 订阅 */
  public void clear() {
    this.consumerMap.clear();
  }

  /**
   * 订阅指定topic + 指定key 的消息
   *
   * @param topic kafka topic
   * @param key key
   * @param consumer 订阅器
   */
  public void subTopic(String topic, String key, Consumer<byte[]> consumer) {
    if (KafkaConsumer.kafkaConsumer == null) {
      return;
    }
    String topicWithKey = ObjectUtils.isEmpty(key) ? topic : (topic + key);
    KafkaConsumer.kafkaConsumer.consumerMap.putIfAbsent(topicWithKey, new CopyOnWriteArraySet<>());
    KafkaConsumer.kafkaConsumer.consumerMap.get(topicWithKey).add(consumer);
  }

  /**
   * 取消订阅指定topic的消息
   *
   * @param topic kafka topic
   * @param consumer 订阅器
   */
  public static void unSubTopic(String topic, Consumer<byte[]> consumer) {
    unSubTopic(topic, null, consumer);
  }

  /**
   * 取消订阅指定topic + 指定key 的消息
   *
   * @param topic kafka topic
   * @param key key
   * @param consumer 订阅器
   */
  public static void unSubTopic(String topic, String key, Consumer<byte[]> consumer) {
    if (KafkaConsumer.kafkaConsumer == null) {
      return;
    }
    String topicWithKey = ObjectUtils.isEmpty(key) ? topic : (topic + key);
    Set<Consumer<byte[]>> set = KafkaConsumer.kafkaConsumer.consumerMap.get(topicWithKey);
    if (set != null) {
      set.remove(consumer);
      if (set.isEmpty()) {
        KafkaConsumer.kafkaConsumer.consumerMap.remove(topicWithKey);
      }
    }
  }

  /**
   * 消费 Acceptor Input数据。
   *
   * @param message Kafka消息
   */
  @KafkaListener(topics = {"#{T(club.vann.kafka.mock.kafka.KafkaTopicFactory).ACCEPTOR_INPUT}"})
  public void onAcceptorInput(ConsumerRecord<String, byte[]> message) {
    onMessage(message);
  }

  /**
   * 消费 Acceptor Output数据。
   *
   * @param message Kafka消息
   */
  @KafkaListener(topics = {"#{T(club.vann.kafka.mock.kafka.KafkaTopicFactory).ACCEPTOR_OUTPUT}"})
  public void onAcceptorOutput(ConsumerRecord<String, byte[]> message) {
    onMessage(message);
  }

  /**
   * 消费市场价格数据。
   *
   * @param message Kafka消息
   */
  @KafkaListener(
      topics = "#{T(club.vann.kafka.mock.kafka.KafkaTopicFactory).MARKET_PRICE_SUMMARY_OUTPUT}")
  public void onMarketPrice(ConsumerRecord<String, byte[]> message) {
    onMessage(message);
  }

  /**
   * 报价操作数据。
   *
   * @param message Kafka消息
   */
  @KafkaListener(topics = "#{T(club.vann.kafka.mock.kafka.KafkaTopicFactory).QUOTE_ACTION_OUTPUT}")
  public void onQuoteActionResult(ConsumerRecord<String, byte[]> message) {
    onMessage(message);
  }

  /**
   * 消费波动率数据。
   *
   * @param message Kafka消息
   */
  @KafkaListener(
      topics = "#{T(club.vann.kafka.mock.kafka.KafkaTopicFactory).REALIZED_VOLATILITY_INPUT}")
  public void onRealizedVolatility(ConsumerRecord<String, byte[]> message) {
    onMessage(message);
  }

  /**
   * 消费自定义成交数据。
   *
   * @param message Kafka消息
   */
  @KafkaListener(topics = "#{T(club.vann.kafka.mock.kafka.KafkaTopicFactory).TRADE_EVENT_OUTPUT}")
  public void onOrderEvent(ConsumerRecord<String, byte[]> message) {
    onMessage(message);
  }

  /**
   * 消费动态表格数据。
   *
   * @param message Kafka消息
   */
  @KafkaListener(
      topicPattern =
          "#{T(club.vann.kafka.mock.kafka.KafkaTopicFactory).TABLE_DATA_OUTPUT.concat('.*')}")
  public void onTableData(ConsumerRecord<String, byte[]> message) {
    onMessage(message);
  }

  private void onMessage(ConsumerRecord<String, byte[]> message, boolean filterByKey) {
    try {
      String topic = message.topic();
      log.debug("Consumer Topic: [{}]", topic);
      String key = message.key();
      byte[] value = message.value();
      String subscribeKey;
      if (key != null && filterByKey) {
        subscribeKey = String.format("%s%s", topic, key);
      } else {
        subscribeKey = topic;
      }
      if (this.consumerMap.containsKey(subscribeKey)) {
        Set<Consumer<byte[]>> set = this.consumerMap.get(subscribeKey);
        if (set != null) {
          set.forEach(
              (Consumer<byte[]> consumer) -> {
                try {
                  consumer.accept(value);
                } catch (Exception e) {
                  log.error(e.getMessage());
                }
              });
        }
      }
    } catch (Exception e) {
      log.error(e.getMessage());
    }
  }

  private void onMessage(ConsumerRecord<String, byte[]> message) {
    onMessage(message, false);
  }
}
